package com.atguigu.edu.realtime220815.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime220815.bean.TradeCourseOrderBean;
import com.atguigu.edu.realtime220815.util.ClickHouseUtils;
import com.atguigu.edu.realtime220815.util.DateFormatUtil;
import com.atguigu.edu.realtime220815.util.KafkaUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.Objects;

  //交易域课程粒度下单各窗口汇总表
  //从 Kafka 读取订单明细数据，过滤 null 数据并按照唯一键对数据去重，
  //统计各课程各窗口下单人数，订单数和订单金额，将数据写入ClickHouse 交易域省份粒度下单各窗口汇总表。


public class DwsTradeCourseOrderWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();


        //TODO 读取Kafka消息
        String topic="dwd_trade_order_detail";
        String groupId="dws_trade_course_order_group";
        FlinkKafkaConsumer<String> kafkaConsumer = KafkaUtils.getFlinkKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaDS = env.addSource(kafkaConsumer);

        // TODO 过滤 null 数据
        SingleOutputStreamOperator<String> filteredStream = kafkaDS.filter(Objects::nonNull);

        // TODO 按照 order_detail_id 分组
        KeyedStream<String, String> keyedStream = filteredStream.keyBy((KeySelector<String, String>) jsonStr -> {
            JSONObject jsonObj = JSONObject.parseObject(jsonStr);
            return jsonObj.getString("id");
        });

        // TODO 去重并转换数据结构
        SingleOutputStreamOperator<TradeCourseOrderBean> processedStream = keyedStream.process(
                new KeyedProcessFunction<String, String, TradeCourseOrderBean>() {

                    private ValueState<JSONObject> lastValueState;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        super.open(parameters);
                        lastValueState = getRuntimeContext().getState(
                                new ValueStateDescriptor<>("last_value_state", JSONObject.class)
                        );
                    }

                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<TradeCourseOrderBean> out) throws Exception {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        JSONObject lastValue = lastValueState.value();
                        if (lastValue != null) {
                            String provinceId = lastValue.getString("province_id");
                            String orderId = lastValue.getString("order_id");
                            String courseId = jsonObj.getString("course_id");
                            String courseName = jsonObj.getString("course_name");
                            String userId = jsonObj.getString("user_id");
                            BigDecimal originalAmount = new BigDecimal(
                                    "-" + lastValue.getString("original_amount"));
                            BigDecimal couponReduce = new BigDecimal(
                                    "-" + lastValue.getString("coupon_reduce"));
                            BigDecimal orderAmount = new BigDecimal(
                                    "-" + lastValue.getString("final_amount"));
                            Long ts = lastValue.getLong("ts") * 1000L;

                            out.collect(TradeCourseOrderBean.builder()
                                    .provinceId(provinceId)
                                    .orderId(orderId)
                                    .courseId(courseId)
                                    .courseName(courseName)
                                    .userId(userId)
                                    .originalAmount(originalAmount)
                                    .couponReduce(couponReduce)
                                    .orderAmount(orderAmount)
                                    .ts(ts)
                                    .build());
                        }
                        lastValueState.update(jsonObj);
                        String provinceId = jsonObj.getString("province_id");
                        String orderId = jsonObj.getString("order_id");
                        String courseId = jsonObj.getString("course_id");
                        String courseName = jsonObj.getString("course_name");
                        String userId = jsonObj.getString("user_id");
                        BigDecimal originalAmount = new BigDecimal(
                                jsonObj.getString("original_amount"));
                        BigDecimal couponReduce = new BigDecimal(
                                jsonObj.getString("coupon_reduce"));
                        BigDecimal orderAmount = new BigDecimal(
                                jsonObj.getString("final_amount"));
                        Long ts = jsonObj.getLong("ts") * 1000L;

                        out.collect(TradeCourseOrderBean.builder()
                                .provinceId(provinceId)
                                .orderId(orderId)
                                .courseId(courseId)
                                .courseName(courseName)
                                .userId(userId)
                                .originalAmount(originalAmount)
                                .couponReduce(couponReduce)
                                .orderAmount(orderAmount)
                                .ts(ts)
                                .build());
                    }
                }
        );
        processedStream.print("***");
        // TODO  按照 order_id 分组
        //KeyedStream<JSONObject, String> keyedByUserIdStream = withWatermarkStream.keyBy(r -> r.getString("user_id"));
        KeyedStream<TradeCourseOrderBean, String> orderIdKeyedStream = processedStream.keyBy(TradeCourseOrderBean::getOrderId);

        // TODO 统计订单数
        SingleOutputStreamOperator<TradeCourseOrderBean> withOrderCountStream = orderIdKeyedStream.process(
                new KeyedProcessFunction<String, TradeCourseOrderBean, TradeCourseOrderBean>() {

                    ValueState<String> lastOrderIdState;

                    @Override
                    public void open(Configuration paramters) throws Exception {
                        super.open(paramters);
                        ValueStateDescriptor<String> lastOrderIdStateDescriptor =
                                new ValueStateDescriptor<>("last_order_id_state", String.class);
                        lastOrderIdStateDescriptor.enableTimeToLive(
                                StateTtlConfig.newBuilder(Time.seconds(10L)).build()
                        );
                        lastOrderIdState = getRuntimeContext().getState(lastOrderIdStateDescriptor);
                    }

                    @Override
                    public void processElement(TradeCourseOrderBean javaBean, Context context, Collector<TradeCourseOrderBean> out) throws Exception {
                        String orderId = javaBean.getOrderId();
                        String lastOrderId = lastOrderIdState.value();
                        if (lastOrderId == null) {
                            javaBean.setOrderCount(1L);
                            out.collect(javaBean);
                            lastOrderIdState.update(orderId);
                        } else {
                            javaBean.setOrderCount(0L);
                            out.collect(javaBean);
                        }
                    }
                }
        );
//        withOrderCountStream.print(">>>");
        // TODO  按照 user_id 分组
        KeyedStream<TradeCourseOrderBean, String> userIdKeyedStream = withOrderCountStream.keyBy(TradeCourseOrderBean::getUserId);

        // TODO  统计用户数
        SingleOutputStreamOperator<TradeCourseOrderBean> withUserCountStream = userIdKeyedStream.process(
                new KeyedProcessFunction<String, TradeCourseOrderBean, TradeCourseOrderBean>() {

                    ValueState<String> lastUserIdDateState;

                    @Override
                    public void open(Configuration paramters) throws Exception {
                        super.open(paramters);
                        ValueStateDescriptor<String> lastOrderIdStateDescriptor =
                                new ValueStateDescriptor<>("last_user_id_state", String.class);
                        lastOrderIdStateDescriptor.enableTimeToLive(
                                StateTtlConfig.newBuilder(Time.days(1)).build()
                        );
                        lastUserIdDateState = getRuntimeContext().getState(lastOrderIdStateDescriptor);
                    }

                    @Override
                    public void processElement(TradeCourseOrderBean javaBean, Context context, Collector<TradeCourseOrderBean> out) throws Exception {
                        String userId = javaBean.getUserId();
                        String currentDate = DateFormatUtil.toDate(javaBean.getTs());
                        String lastOrderIdDate = lastUserIdDateState.value();
                        if (StringUtils.isEmpty(lastOrderIdDate) ||!lastOrderIdDate.equals(currentDate)) {
                            javaBean.setUserCount(1L);
                            out.collect(javaBean);
                            lastUserIdDateState.update(currentDate);
                        } else {
                            javaBean.setUserCount(0L);
                            out.collect(javaBean);
                        }
                    }
                }
        );
//        withUserCountStream.print("???");
        // TODO  设置水位线
        SingleOutputStreamOperator<TradeCourseOrderBean> withWatermarkStream = withUserCountStream.assignTimestampsAndWatermarks(
                WatermarkStrategy.<TradeCourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(((element, recordTimestamp) -> element.getTs()))
        );

        // TODO  按照课程 ID 分组
        KeyedStream<TradeCourseOrderBean, String> keyedByCourseIdStream =withWatermarkStream.keyBy(TradeCourseOrderBean::getCourseId);

        // TODO  开窗
        WindowedStream<TradeCourseOrderBean, String, TimeWindow> windowDS = keyedByCourseIdStream
                .window(TumblingEventTimeWindows.of(org.apache.flink.streaming.api.windowing.time.Time.seconds(10L)))
                .allowedLateness(org.apache.flink.streaming.api.windowing.time.Time.seconds(5L));

        // TODO 聚合
        SingleOutputStreamOperator<TradeCourseOrderBean> reduceDS = windowDS.reduce(
                (ReduceFunction<TradeCourseOrderBean>) (value1, value2) -> {
                    value1.setOrderCount(value1.getOrderCount() + value2.getOrderCount());
                    value1.setUserCount(value1.getUserCount() + value2.getUserCount());
                    value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                    value1.setCouponReduce(value1.getCouponReduce().add(value2.getCouponReduce()));
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));

                    return value1;
                },
                (WindowFunction<TradeCourseOrderBean, TradeCourseOrderBean, String, TimeWindow>) (s, window, input, out) -> {
                    for (TradeCourseOrderBean tradeCourseOrderBean : input) {
                        tradeCourseOrderBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        tradeCourseOrderBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        out.collect(tradeCourseOrderBean);
                    }
                }
        );
//        reduceDS.print(">>>");



        //TODO 将关联的结果写入ck
//        withCategoryNameDS.print(">>>");

        reduceDS.addSink(ClickHouseUtils.getClickHouseSinkFunction("insert into dws_trade_course_order_window" +
                " values (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
